package com.wl.es7.study.test.client.java;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HeaderElement;
import org.apache.http.HeaderElementIterator;
import org.apache.http.HttpHost;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.message.BasicHeaderElementIterator;
import org.apache.http.protocol.HTTP;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;

import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

@Slf4j
public class ElasticSearchJavaUtils {

    private static final ObjectMapper OBJECT_MAPPER;


    // 15分钟 (毫秒)
    private static final long KEEP_ALIVE_MS = 15 * 60 * 1000;

    static {
        OBJECT_MAPPER = new ObjectMapper();
        OBJECT_MAPPER.setSerializationInclusion(JsonInclude.Include.NON_NULL);
    }
    public static RestHighLevelClient createClient(){
       return new RestHighLevelClient(
               RestClient.builder(new HttpHost("192.168.43.100", 9200)));
    }

    @SneakyThrows
    public static RestHighLevelClient createClientPool(){

        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
                .setIoThreadCount(Runtime.getRuntime().availableProcessors())
                .setConnectTimeout(10)
                .setRcvBufSize(5)
                .setSoKeepAlive(true)
                .build();
        PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(new DefaultConnectingIOReactor(ioReactorConfig));
        connManager.setMaxTotal(100);
        connManager.setDefaultMaxPerRoute(100);

        // CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        // credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));

        RestClientBuilder restClientBuilder = RestClient.builder(new HttpHost("192.168.43.100", 9200))
                .setHttpClientConfigCallback(httpClientBuilder -> {
                    // 连接保持
                    httpClientBuilder.setKeepAliveStrategy((response, context) -> {
                        final HeaderElementIterator it = new BasicHeaderElementIterator(
                                response.headerIterator(HTTP.CONN_KEEP_ALIVE));
                        while (it.hasNext()) {
                            final HeaderElement he = it.nextElement();
                            final String param = he.getName();
                            final String value = he.getValue();
                            if (value != null && param.equalsIgnoreCase("timeout")) {
                                try {
                                    return Long.parseLong(value) * 1000;
                                } catch (final NumberFormatException ignore) {
                                }
                            }
                        }
                        return KEEP_ALIVE_MS;
                    });
                    // 最大连接数
                    httpClientBuilder.setMaxConnTotal(100);
                    // 最大路由连接数
                    httpClientBuilder.setMaxConnPerRoute(100);
                    // httpClientBuilder.disableAuthCaching();
                    // httpClientBuilder.setDefaultCredentialsProvider();
                    httpClientBuilder.setConnectionManager(connManager);
                    return httpClientBuilder;
                })
                .setRequestConfigCallback(requestConfigBuilder -> {
                    // 连接目标url的连接超时时间（毫秒）
                    requestConfigBuilder.setConnectTimeout(5000);
                    // 请求获取数据的超时时间(即响应时间)，单位毫秒
                    requestConfigBuilder.setSocketTimeout(40000);
                    // 从connect Manager(连接池)获取Connection 超时时间，单位毫秒
                    requestConfigBuilder.setConnectionRequestTimeout(1000);
                    return requestConfigBuilder;
                });

        return new RestHighLevelClient(restClientBuilder);
    }


    public static void execute(Consumer<RestHighLevelClient> task){
        try {
            RestHighLevelClient client = createClient();
            task.accept(client);
            client.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @SneakyThrows
    public static <T> T parseObject(String json, Class<T> clazz){
        return OBJECT_MAPPER.readValue(json,clazz);
    }

    @SneakyThrows
    public static String ObjectToString(Object obj){
        return OBJECT_MAPPER.writeValueAsString(obj);
    }
}
